package com.hutu.producter;

import com.hutu.model.Student;
import jakarta.annotation.Resource;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.header.Headers;
import org.apache.kafka.common.header.internals.RecordHeaders;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

@Service
public class KafkaProducerService {
    @Resource
    private KafkaTemplate<String, Object> kafkaTemplate;
/*    @Resource
    private KafkaTemplate<String, Student> studentKafkaTemplate;*/


    public void sendMessage() {
        kafkaTemplate.send("kafka-topic","test send msg");
    }


    public void sendMessageStudent(Student studentJson) {
        kafkaTemplate.send("test-kafka-user",studentJson);
    }

    public void sendMessage2() {
        Message<String> message = MessageBuilder
                // 信息体
                .withPayload("hello kafka")
                // 设置主题
                .setHeader(KafkaHeaders.TOPIC, "test-kafka")
                .build();
        kafkaTemplate.send(message);
    }

    public void sendMessage3() {
        Headers headers = new RecordHeaders();
        headers.add("test-header", "test-header-value".getBytes());
//        如果没有指定分区,则按照key的hash值来确定分区,如果key的hash值一致则会放入到相同分区中
        ProducerRecord<String,Object> record = new ProducerRecord<>(
                "test-kafka",
                0,
                System.currentTimeMillis(),
                "key",
                "hello kafka ProducerRecord",
                headers
        );
        kafkaTemplate.send(record);
    }

}
